package com.cmic.plugin.reader.spdmysqlreader;

import com.alibaba.datax.common.base.IdRecord;
import com.alibaba.datax.common.constant.IdRecordConstants;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.element.StringColumn;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.common.util.RedissonUtils;
import com.alibaba.datax.plugin.rdbms.util.DBUtil;
import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.sql.*;
import java.sql.Date;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

public class SpdMysqlReader extends Reader {

    private static final DataBaseType DATABASE_TYPE = DataBaseType.MySql;

    public static class Job extends Reader.Job {
        private static final Logger LOG = LoggerFactory
                .getLogger(Job.class);

        private Configuration originalConfig = null;

        @Override
        public void init() {
            this.originalConfig = super.getPluginJobConf();
            LOG.info("job.init:{}",originalConfig);
        }

        @Override
        public void destroy() {

        }

        @Override
        public List<Configuration> split(int adviceNumber) {
            List<Configuration> configurations = new ArrayList<>();
            configurations.add(this.originalConfig);
            return configurations;
        }
    }


    public static class Task extends Reader.Task {

        private static final Logger LOG = LoggerFactory
                .getLogger(SpdMysqlReader.Task.class);
        //配置文件
        private Configuration readerSliceConfig;
        private Connection conn ;
        private RedissonUtils redissonUtils;
        private RTopic idRecord;

        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
        @Override
        public void init() {
            this.readerSliceConfig = super.getPluginJobConf();
            LOG.info("task.init:{}",readerSliceConfig);

            String username = readerSliceConfig.getString("username");
            String password = readerSliceConfig.getString("password");
            String jdbcUrl = readerSliceConfig.getString("jdbcUrl");

            LOG.info("username:{}",username);


            conn = DBUtil.getConnection(DATABASE_TYPE, jdbcUrl,
                    username, password);

            String redisHost = System.getProperty("redisHost");
            String redisPasswd = System.getProperty("redisPasswd");
            String redisDatabase = System.getProperty("redisDatabase");
            redissonUtils = new RedissonUtils(redisHost, redisPasswd, 6379, Integer.parseInt(redisDatabase));
            idRecord = redissonUtils.getRedissonClient().getTopic("idRecord");

        }

        @Override
        public void destroy() {

        }

        @Override
        public void startRead(RecordSender recordSender) throws SQLException, ParseException {
            String querySql = readerSliceConfig.getString("querySql");
            String lastVal = System.getProperty("lastVal");
            String incrementField = readerSliceConfig.getString("incrementField");



            List<Object> columns = readerSliceConfig.getList("column");

            String taskId = System.getProperty("taskId");
            LOG.info("开始写入数据任务:{}",taskId);

            PreparedStatement preparedStatement = conn.prepareStatement(querySql);
            if(StringUtils.isNotBlank(lastVal)){
                preparedStatement.setDate(1,new Date(sdf.parse(lastVal).getTime()));
            }

            preparedStatement.setFetchSize(Integer.MIN_VALUE);
            ResultSet resultSet = preparedStatement.executeQuery();

            ResultSetMetaData metaData = resultSet.getMetaData();
            while (resultSet.next()){
                Record record = recordSender.createRecord();
                lastVal = sdf.format(resultSet.getDate(incrementField));
                Map<String,String> meta = new LinkedHashMap<>();
                meta.put(incrementField,lastVal);
                record.setMeta(meta);
                try {
                    for (int i=0;i<columns.size();i++) {
                        String columnName = (String)columns.get(i);
                        int columnType = metaData.getColumnType(i+1);
                        switch (columnType){
                            case Types.CHAR:
                            case Types.NCHAR:
                            case Types.VARCHAR:
                            case Types.LONGVARCHAR:
                            case Types.NVARCHAR:
                            case Types.LONGNVARCHAR:
                                String rawData = resultSet.getString(columnName);
                                record.addColumn(new StringColumn(rawData));
                                break;
                            case Types.DATE:
                            case Types.TIMESTAMP:
                            case Types.TIME:
                                Date date = resultSet.getDate(columnName);
                                record.addColumn(new StringColumn(sdf.format(date)));
                                break;
                            case Types.INTEGER:
                            case Types.BIGINT:
                                int num = resultSet.getInt(columnName);
                                record.addColumn(new StringColumn(String.valueOf(num)));
                                break;
                            case Types.DECIMAL:
                                BigDecimal decimal = resultSet.getBigDecimal(columnName);
                                if(decimal==null){
                                    record.addColumn(new StringColumn("-"));
                                }else{
                                    record.addColumn(new StringColumn(decimal.toPlainString()));

                                }
                                break;
                            default:
                                LOG.error("无效数据类型");
                                StringColumn stringColumn = new StringColumn("-");
                                record.addColumn(stringColumn);
                        }
                    }

                } catch (Exception e) {
                    throw DataXException.asDataXException(SpdMysqlReaderErrorCode.ERROR,"构造一个record失败.", e);
                }
                recordSender.sendToWriter(record);
                idRecord.publish(JSONObject.toJSONString(new IdRecord(taskId,lastVal, IdRecordConstants.READ)));
            }
            DBUtil.closeDBResources(preparedStatement, conn);
            redissonUtils.close();
        }
    }


}
